package com.xj.hadoop.hive;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.ql.exec.FileSinkOperator.RecordWriter;
import org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.JobConfigurable;
import org.apache.hadoop.util.Progressable;

import java.io.IOException;
import java.util.Properties;

public class HiveOutputformat<K extends WritableComparable, V extends Writable> extends
		HiveIgnoreKeyTextOutputFormat<K, V> {
	public HiveOutputformat() {
		System.out.println("----------HiveOutputFormat init-----------");
	}

	@Override
	public RecordWriter getHiveRecordWriter(JobConf jc, Path finalOutPath,
			Class<? extends Writable> valueClass, boolean isCompressed, Properties tableProperties,
			Progressable progress) throws IOException {
		System.out.println("getHiveRecordWriter.......");
		HiveRecordWriter writer = new HiveRecordWriter(super.getHiveRecordWriter(jc, finalOutPath,
				BytesWritable.class, isCompressed, tableProperties, progress));
		writer.configure(jc);
		return writer;
	}

	@Override
	public org.apache.hadoop.mapred.RecordWriter<K, V> getRecordWriter(FileSystem ignored, JobConf job,
			String name, Progressable progress) throws IOException {
		System.out.println("getRecordWriter.........");
		return null;
	}

	public static class HiveRecordWriter implements RecordWriter, JobConfigurable {
		RecordWriter writer;
		BytesWritable bytesWritable;

		public HiveRecordWriter(RecordWriter writer) {
			this.writer = writer;
			bytesWritable = new BytesWritable();
		}

		@Override
		public void configure(JobConf arg0) {
			System.out.println("outputformat configure null ...");
		}

		@Override
		public void write(Writable w) throws IOException {
			System.out.println("++++++++++++++++++++++++++++++++");
			byte[] input;
			if (w instanceof Text) {
				input = ((Text) w).getBytes();
			} else {
				assert (w instanceof BytesWritable);
				input = ((BytesWritable) w).getBytes();
			}
			bytesWritable.set(input, 0, input.length);
			writer.write(bytesWritable);
		}

		@Override
		public void close(boolean abort) throws IOException {
			writer.close(abort);
		}

	}
}
